package com.example.testkafka.consumer;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.listener.ConsumerSeekAware;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 * 每个带有@KafkaListener的方法都会启动concurrency个消费者
 */
@Component
public class TestConsumer implements ConsumerSeekAware {

//    @Autowired
//    private KafkaListenerContainerFactory kafkaListenerContainerFactory;

    //可消费任意分区
//    @KafkaListener(topics = "test_topic",groupId = "group18",containerFactory = "kafkaListenerContainerFactory")
    //只可消费指定分区
//    @KafkaListener(topicPartitions = {@TopicPartition(topic = "test_topic",partitions = {"1","2"})},groupId = "group18",containerFactory = "kafkaListenerContainerFactory")
    /**
     * 单条消费;非显示提交:支持除MANUAL和MANUAL_IMMEDIATE之外的其他提交方式
     * 单条消费入参需按此顺序按需添加:data(消息), acknowledgment(位移提交对象), consumer(消费者实例)
     * @param record
     */
    public void singleConsumeAutoCommit(ConsumerRecord<?, ?> record) {

        System.out.printf("[消费者]thread:" + Thread.currentThread().getName() + ",groupid:group17,topic = %s,partition = %d, offset = %d, value = %s \n", record.topic(),record.partition(), record.offset(), record.value());
    }

    @KafkaListener(topics = "test_topic",groupId = "group15",containerFactory = "kafkaListenerContainerFactory",id = "helloworld")
    /**
     * 单条消费;显示提交:仅支持MANUAL和MANUAL_IMMEDIATE
     * 单条消费入参需按此顺序按需添加:data(消息), acknowledgment(位移提交对象), consumer(消费者实例)
     * @param record
     * @param ack
     */
    public void singleConsumeManualCommit(ConsumerRecord<?, ?> record, Acknowledgment ack) {

        System.out.printf("[消费者]thread:" + Thread.currentThread().getName() + ",groupid:group2,topic = %s,partition = %d, offset = %d, value = %s \n", record.topic(),record.partition(), record.offset(), record.value());
        ack.acknowledge();//纯手动提交offset
    }

//    @KafkaListener(topics = "test_topic",groupId = "group12",containerFactory = "kafkaListenerContainerFactory")
    /**
     * 批量消费;非显示提交:支持除MANUAL和MANUAL_IMMEDIATE之外的其他提交方式
     * 批量消费入参需按此顺序按需添加:data(消息列表), acknowledgment(位移提交对象), consumer(消费者实例)
     * @param consumerRecords
     */
    public void bachConsumeAutoCommit(List<ConsumerRecord> consumerRecords) {

        Iterator<ConsumerRecord> iterator = consumerRecords.iterator();
        while (iterator.hasNext()) {
            ConsumerRecord record = iterator.next();
            System.out.printf("[消费者]thread:" + Thread.currentThread().getName() + ",groupid:group2,topic = %s,partition = %d, offset = %d, value = %s \n", record.topic(),record.partition(), record.offset(), record.value());
        }
    }

//    @KafkaListener(topics = "test_topic",groupId = "group11",containerFactory = "kafkaListenerContainerFactory")
    /**
     * 批量消费;显示提交:仅支持MANUAL和MANUAL_IMMEDIATE
     * 批量消费入参需按此顺序按需添加:data(消息列表), acknowledgment(位移提交对象), consumer(消费者实例)
     * @param consumerRecords
     * @param ack   只有当AckMode为AckMode.MANUAL_IMMEDIATE或AckMode.MANUAL时，才会有此入参，用于手动提交
     */
    public void consumerListener(List<ConsumerRecord> consumerRecords, Acknowledgment ack) {

        Iterator<ConsumerRecord> iterator = consumerRecords.iterator();
        while (iterator.hasNext()) {
            ConsumerRecord record = iterator.next();
            System.out.printf("[消费者]thread:" + Thread.currentThread().getName() + ",groupid:group2,topic = %s,partition = %d, offset = %d, value = %s \n", record.topic(),record.partition(), record.offset(), record.value());
        }
        ack.acknowledge();//纯手动提交offset
    }

    /**
     *  容器初始化调用
     * @param callback
     */
    @Override
    public void registerSeekCallback(ConsumerSeekCallback callback) {

        //TODO
        System.out.println("hello");
    }

    /**
     * 分区重分配
     * @param assignments
     * @param callback
     */
    @Override
    public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {

        //TODO
        System.out.println("hello");
    }

    @Override
    public void onIdleContainer(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {

        //TODO
        System.out.println("hello");
    }
}
